// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/csv/converter.h>

#include <array>
#include <cstring>
#include <limits>
#include <sstream>
#include <string>
#include <type_traits>
#include <vector>

#include <nebula/array/builder_binary.h>
#include <nebula/array/builder_decimal.h>
#include <nebula/array/builder_dict.h>
#include <nebula/array/builder_primitive.h>
#include <nebula/csv/parser.h>
#include <turbo/utility/status.h>
#include <nebula/types/type.h>
#include <nebula/types/type_fwd.h>
#include <nebula/types/type_traits.h>
#include <turbo/base/checked_cast.h>
#include <nebula/numeric/decimal.h>
#include <nebula/util/trie.h>
#include <nebula/util/utf8_internal.h>
#include <nebula/util/value_parsing.h>  // IWYU pragma: keep

namespace nebula::csv {

    using internal::Trie;
    using internal::TrieBuilder;

    namespace {

        turbo::Status GenericConversionError(const std::shared_ptr<DataType> &type, const uint8_t *data,
                                      uint32_t size) {
            return turbo::invalid_argument_error("CSV conversion error to ", type->to_string(),
                                   ": invalid value '",
                                   std::string(reinterpret_cast<const char *>(data), size), "'");
        }

        inline bool IsWhitespace(uint8_t c) {
            if (TURBO_LIKELY(c > ' ')) {
                return false;
            }
            return c == ' ' || c == '\t';
        }

        // Updates data_inout and size_inout to not include leading/trailing whitespace
        // characters.
        inline void TrimWhiteSpace(const uint8_t **data_inout, uint32_t *size_inout) {
            const uint8_t *&data = *data_inout;
            uint32_t &size = *size_inout;
            // Skip trailing whitespace
            if (TURBO_LIKELY(size > 0) && TURBO_UNLIKELY(IsWhitespace(data[size - 1]))) {
                const uint8_t *p = data + size - 1;
                while (size > 0 && IsWhitespace(*p)) {
                    --size;
                    --p;
                }
            }
            // Skip leading whitespace
            if (TURBO_LIKELY(size > 0) && TURBO_UNLIKELY(IsWhitespace(data[0]))) {
                while (size > 0 && IsWhitespace(*data)) {
                    --size;
                    ++data;
                }
            }
        }

        turbo::Status InitializeTrie(const std::vector<std::string> &inputs, Trie *trie) {
            TrieBuilder builder;
            for (const auto &s: inputs) {
                TURBO_RETURN_NOT_OK(builder.append(s, true /* allow_duplicates */));
            }
            *trie = builder.finish();
            return turbo::OkStatus();
        }

// Presize a builder based on parser contents
        template<typename BuilderType>
        turbo::Status PresizeBuilder(const BlockParser &parser, BuilderType *builder) {
            TURBO_RETURN_NOT_OK(builder->resize(parser.num_rows()));
            if constexpr (is_base_binary_type<typename BuilderType::TypeClass>::value) {
                return builder->reserve_data(parser.num_bytes());
            } else {
                return turbo::OkStatus();
            }
        }

/////////////////////////////////////////////////////////////////////////
// Per-type value decoders

        struct ValueDecoder {
            explicit ValueDecoder(const std::shared_ptr<DataType> &type,
                                  const ConvertOptions &options)
                    : type_(type), options_(options) {}

            turbo::Status Initialize() {
                // TODO no need to build a separate Trie for each instance
                return InitializeTrie(options_.null_values, &null_trie_);
            }

            bool is_null(const uint8_t *data, uint32_t size, bool quoted) {
                if (quoted && !options_.quoted_strings_can_be_null) {
                    return false;
                }
                return null_trie_.Find(std::string_view(reinterpret_cast<const char *>(data), size)) >=
                       0;
            }

        protected:
            Trie null_trie_;
            const std::shared_ptr<DataType> type_;
            const ConvertOptions &options_;
        };

        //
        // Value decoder for fixed-size binary
        //

        struct FixedSizeBinaryValueDecoder : public ValueDecoder {
            using value_type = const uint8_t *;

            explicit FixedSizeBinaryValueDecoder(const std::shared_ptr<DataType> &type,
                                                 const ConvertOptions &options)
                    : ValueDecoder(type, options),
                      byte_width_(turbo::checked_cast<const FixedSizeBinaryType &>(*type).byte_width()) {}

            turbo::Status Decode(const uint8_t *data, uint32_t size, bool quoted, value_type *out) {
                if (TURBO_UNLIKELY(size != byte_width_)) {
                    return turbo::invalid_argument_error("CSV conversion error to ", type_->to_string(), ": got a ",
                                           size, "-byte long string");
                }
                *out = data;
                return turbo::OkStatus();
            }

        protected:
            const uint32_t byte_width_;
        };

//
// Value decoder for variable-size binary
//

        template<bool CheckUTF8>
        struct BinaryValueDecoder : public ValueDecoder {
            using value_type = std::string_view;

            using ValueDecoder::ValueDecoder;

            turbo::Status Initialize() {
                util::InitializeUTF8();
                return ValueDecoder::Initialize();
            }

            turbo::Status Decode(const uint8_t *data, uint32_t size, bool quoted, value_type *out) {
                if (CheckUTF8 && TURBO_UNLIKELY(!util::ValidateUTF8Inline(data, size))) {
                    return turbo::invalid_argument_error("CSV conversion error to ", type_->to_string(),
                                           ": invalid UTF8 data");
                }
                *out = {reinterpret_cast<const char *>(data), size};
                return turbo::OkStatus();
            }

            bool is_null(const uint8_t *data, uint32_t size, bool quoted) {
                return options_.strings_can_be_null &&
                       (!quoted || options_.quoted_strings_can_be_null) &&
                       ValueDecoder::is_null(data, size, false /* quoted */);
            }
        };

//
// Value decoder for integers, floats and temporals
//

        template<typename T>
        static nebula::internal::StringConverter<T> MakeStringConverter(
                const ConvertOptions &options) {
            if constexpr (is_floating_type<T>::value) {
                return nebula::internal::StringConverter<T>{options.decimal_point};
            } else {
                return nebula::internal::StringConverter<T>{};
            }
        }

        template<typename T>
        struct NumericValueDecoder : public ValueDecoder {
            using value_type = typename T::c_type;

            NumericValueDecoder(const std::shared_ptr<DataType> &type,
                                const ConvertOptions &options)
                    : ValueDecoder(type, options),
                      concrete_type_(turbo::checked_cast<const T &>(*type)),
                      string_converter_(MakeStringConverter<T>(options)) {}

            turbo::Status Decode(const uint8_t *data, uint32_t size, bool quoted, value_type *out) {
                // XXX should quoted values be allowed at all?
                TrimWhiteSpace(&data, &size);
                if (TURBO_UNLIKELY(!string_converter_.Convert(
                        concrete_type_, reinterpret_cast<const char *>(data), size, out))) {
                    return GenericConversionError(type_, data, size);
                }
                return turbo::OkStatus();
            }

        protected:
            const T &concrete_type_;
            nebula::internal::StringConverter<T> string_converter_;
        };

//
// Value decoder for booleans
//

        struct BooleanValueDecoder : public ValueDecoder {
            using value_type = bool;

            using ValueDecoder::ValueDecoder;

            turbo::Status Initialize() {
                // TODO no need to build separate Tries for each instance
                TURBO_RETURN_NOT_OK(InitializeTrie(options_.true_values, &true_trie_));
                TURBO_RETURN_NOT_OK(InitializeTrie(options_.false_values, &false_trie_));
                return ValueDecoder::Initialize();
            }

            turbo::Status Decode(const uint8_t *data, uint32_t size, bool quoted, value_type *out) {
                // XXX should quoted values be allowed at all?
                if (false_trie_.Find(std::string_view(reinterpret_cast<const char *>(data), size)) >=
                    0) {
                    *out = false;
                    return turbo::OkStatus();
                }
                if (TURBO_LIKELY(true_trie_.Find(std::string_view(
                        reinterpret_cast<const char *>(data), size)) >= 0)) {
                    *out = true;
                    return turbo::OkStatus();
                }
                return GenericConversionError(type_, data, size);
            }

        protected:
            Trie true_trie_;
            Trie false_trie_;
        };

//
// Value decoder for decimals
//

        struct DecimalValueDecoder : public ValueDecoder {
            using value_type = Decimal128;

            explicit DecimalValueDecoder(const std::shared_ptr<DataType> &type,
                                         const ConvertOptions &options)
                    : ValueDecoder(type, options),
                      decimal_type_(turbo::checked_cast<const DecimalType &>(*type_)),
                      type_precision_(decimal_type_.precision()),
                      type_scale_(decimal_type_.scale()) {}

            turbo::Status Decode(const uint8_t *data, uint32_t size, bool quoted, value_type *out) {
                TrimWhiteSpace(&data, &size);
                Decimal128 decimal;
                int32_t precision, scale;
                std::string_view view(reinterpret_cast<const char *>(data), size);
                TURBO_RETURN_NOT_OK(Decimal128::from_string(view, &decimal, &precision, &scale));
                if (precision > type_precision_) {
                    return turbo::invalid_argument_error("Error converting '", view, "' to ", type_->to_string(),
                                           ": precision not supported by type.");
                }
                if (scale != type_scale_) {
                    TURBO_MOVE_OR_RAISE(*out, decimal.Rescale(scale, type_scale_));
                } else {
                    *out = std::move(decimal);
                }
                return turbo::OkStatus();
            }

        protected:
            const DecimalType &decimal_type_;
            const int32_t type_precision_;
            const int32_t type_scale_;
        };

//
// Value decoder wrapper for decimals with a non-default decimal point
//

        template<typename WrappedDecoder>
        struct CustomDecimalPointValueDecoder : public ValueDecoder {
            using value_type = typename WrappedDecoder::value_type;

            explicit CustomDecimalPointValueDecoder(const std::shared_ptr<DataType> &type,
                                                    const ConvertOptions &options)
                    : ValueDecoder(type, options), wrapped_decoder_(type, options) {}

            turbo::Status Initialize() {
                TURBO_RETURN_NOT_OK(wrapped_decoder_.Initialize());
                for (int i = 0; i < 256; ++i) {
                    mapping_[i] = i;
                }
                mapping_[options_.decimal_point] = '.';
                mapping_['.'] = options_.decimal_point;  // error out on standard decimal point
                temp_.resize(30);
                return turbo::OkStatus();
            }

            turbo::Status Decode(const uint8_t *data, uint32_t size, bool quoted, value_type *out) {
                if (TURBO_UNLIKELY(size > temp_.size())) {
                    temp_.resize(size);
                }
                uint8_t *temp_data = temp_.data();
                for (uint32_t i = 0; i < size; ++i) {
                    temp_data[i] = mapping_[data[i]];
                }
                if (TURBO_UNLIKELY(
                        !wrapped_decoder_.Decode(temp_data, size, quoted, out).ok())) {
                    return GenericConversionError(type_, data, size);
                }
                return turbo::OkStatus();
            }

            bool is_null(const uint8_t *data, uint32_t size, bool quoted) {
                return wrapped_decoder_.is_null(data, size, quoted);
            }

        protected:
            WrappedDecoder wrapped_decoder_;
            std::array<uint8_t, 256> mapping_;
            std::vector<uint8_t> temp_;
        };

//
// Value decoders for timestamps
//

        struct InlineISO8601ValueDecoder : public ValueDecoder {
            using value_type = int64_t;

            explicit InlineISO8601ValueDecoder(const std::shared_ptr<DataType> &type,
                                               const ConvertOptions &options)
                    : ValueDecoder(type, options),
                      unit_(turbo::checked_cast<const TimestampType &>(*type_).unit()),
                      expect_timezone_(!turbo::checked_cast<const TimestampType &>(*type_).timezone().empty()) {
            }

            turbo::Status Decode(const uint8_t *data, uint32_t size, bool quoted, value_type *out) {
                bool zone_offset_present = false;
                if (TURBO_UNLIKELY(
                        !internal::ParseTimestampISO8601(reinterpret_cast<const char *>(data), size,
                                                         unit_, out, &zone_offset_present))) {
                    return GenericConversionError(type_, data, size);
                }
                if (zone_offset_present != expect_timezone_) {
                    if (expect_timezone_) {
                        return turbo::invalid_argument_error("CSV conversion error to ", type_->to_string(),
                                               ": expected a zone offset in '",
                                               std::string(reinterpret_cast<const char *>(data), size),
                                               "'. If these timestamps are in local time, parse them as "
                                               "timestamps without timezone, then call assume_timezone.");
                    } else {
                        return turbo::invalid_argument_error("CSV conversion error to ", type_->to_string(),
                                               ": expected no zone offset in '",
                                               std::string(reinterpret_cast<const char *>(data), size),
                                               "'");
                    }
                }
                return turbo::OkStatus();
            }

        protected:
            TimeUnit::type unit_;
            bool expect_timezone_;
        };

        struct SingleParserTimestampValueDecoder : public ValueDecoder {
            using value_type = int64_t;

            explicit SingleParserTimestampValueDecoder(const std::shared_ptr<DataType> &type,
                                                       const ConvertOptions &options)
                    : ValueDecoder(type, options),
                      unit_(turbo::checked_cast<const TimestampType &>(*type_).unit()),
                      expect_timezone_(!turbo::checked_cast<const TimestampType &>(*type_).timezone().empty()),
                      parser_(*options_.timestamp_parsers[0]) {}

            turbo::Status Decode(const uint8_t *data, uint32_t size, bool quoted, value_type *out) {
                bool zone_offset_present = false;
                if (TURBO_UNLIKELY(!parser_(reinterpret_cast<const char *>(data), size, unit_,
                                            out, &zone_offset_present))) {
                    return GenericConversionError(type_, data, size);
                }
                if (zone_offset_present != expect_timezone_) {
                    if (expect_timezone_) {
                        return turbo::invalid_argument_error("CSV conversion error to ", type_->to_string(),
                                               ": expected a zone offset in '",
                                               std::string(reinterpret_cast<const char *>(data), size),
                                               "'. If these timestamps are in local time, parse them as "
                                               "timestamps without timezone, then call assume_timezone. "
                                               "If using strptime, ensure '%z' is in the format string.");
                    } else {
                        return turbo::invalid_argument_error("CSV conversion error to ", type_->to_string(),
                                               ": expected no zone offset in '",
                                               std::string(reinterpret_cast<const char *>(data), size),
                                               "'");
                    }
                }
                return turbo::OkStatus();
            }

        protected:
            TimeUnit::type unit_;
            bool expect_timezone_;
            const TimestampParser &parser_;
        };

        struct MultipleParsersTimestampValueDecoder : public ValueDecoder {
            using value_type = int64_t;

            explicit MultipleParsersTimestampValueDecoder(const std::shared_ptr<DataType> &type,
                                                          const ConvertOptions &options)
                    : ValueDecoder(type, options),
                      unit_(turbo::checked_cast<const TimestampType &>(*type_).unit()),
                      expect_timezone_(!turbo::checked_cast<const TimestampType &>(*type_).timezone().empty()),
                      parsers_(GetParsers(options_)) {}

            turbo::Status Decode(const uint8_t *data, uint32_t size, bool quoted, value_type *out) {
                bool zone_offset_present = false;
                for (const auto &parser: parsers_) {
                    if (parser->operator()(reinterpret_cast<const char *>(data), size, unit_, out,
                                           &zone_offset_present) &&
                        zone_offset_present == expect_timezone_) {
                        return turbo::OkStatus();
                    }
                }
                return GenericConversionError(type_, data, size);
            }

        protected:
            using ParserVector = std::vector<const TimestampParser *>;

            static ParserVector GetParsers(const ConvertOptions &options) {
                ParserVector parsers(options.timestamp_parsers.size());
                for (size_t i = 0; i < options.timestamp_parsers.size(); ++i) {
                    parsers[i] = options.timestamp_parsers[i].get();
                }
                return parsers;
            }

            TimeUnit::type unit_;
            bool expect_timezone_;
            std::vector<const TimestampParser *> parsers_;
        };

/////////////////////////////////////////////////////////////////////////
// Concrete Converter hierarchy

        class ConcreteConverter : public Converter {
        public:
            using Converter::Converter;
        };

        class ConcreteDictionaryConverter : public DictionaryConverter {
        public:
            using DictionaryConverter::DictionaryConverter;
        };

//
// Concrete Converter for nulls
//

        class NullConverter : public ConcreteConverter {
        public:
            NullConverter(const std::shared_ptr<DataType> &type, const ConvertOptions &options,
                          MemoryPool *pool)
                    : ConcreteConverter(type, options, pool), decoder_(type_, options_) {}

            turbo::Result<std::shared_ptr<Array>> Convert(const BlockParser &parser,
                                                   int32_t col_index) override {
                NullBuilder builder(pool_);

                auto visit = [&](const uint8_t *data, uint32_t size, bool quoted) -> turbo::Status {
                    if (TURBO_LIKELY(decoder_.is_null(data, size, quoted))) {
                        return builder.append_null();
                    } else {
                        return GenericConversionError(type_, data, size);
                    }
                };
                TURBO_RETURN_NOT_OK(parser.visit_column(col_index, visit));
                std::shared_ptr<Array> res;
                TURBO_RETURN_NOT_OK(builder.finish(&res));
                return res;
            }

        protected:
            turbo::Status Initialize() override { return decoder_.Initialize(); }

            ValueDecoder decoder_;
        };

//
// Concrete Converter for primitives
//

        template<typename T, typename ValueDecoderType>
        class PrimitiveConverter : public ConcreteConverter {
        public:
            PrimitiveConverter(const std::shared_ptr<DataType> &type, const ConvertOptions &options,
                               MemoryPool *pool)
                    : ConcreteConverter(type, options, pool), decoder_(type_, options_) {}

            turbo::Result<std::shared_ptr<Array>> Convert(const BlockParser &parser,
                                                   int32_t col_index) override {
                using BuilderType = typename TypeTraits<T>::BuilderType;
                using value_type = typename ValueDecoderType::value_type;

                BuilderType builder(type_, pool_);
                TURBO_RETURN_NOT_OK(PresizeBuilder(parser, &builder));

                auto visit = [&](const uint8_t *data, uint32_t size, bool quoted) -> turbo::Status {
                    if (decoder_.is_null(data, size, quoted /* quoted */)) {
                        return builder.append_null();
                    }
                    value_type value{};
                    TURBO_RETURN_NOT_OK(decoder_.Decode(data, size, quoted, &value));
                    builder.unsafe_append(value);
                    return turbo::OkStatus();
                };
                TURBO_RETURN_NOT_OK(parser.visit_column(col_index, visit));

                std::shared_ptr<Array> res;
                TURBO_RETURN_NOT_OK(builder.finish(&res));
                return res;
            }

        protected:
            turbo::Status Initialize() override { return decoder_.Initialize(); }

            ValueDecoderType decoder_;
        };

//
// Concrete Converter for dictionaries
//

        template<typename T, typename ValueDecoderType>
        class TypedDictionaryConverter : public ConcreteDictionaryConverter {
        public:
            TypedDictionaryConverter(const std::shared_ptr<DataType> &value_type,
                                     const ConvertOptions &options, MemoryPool *pool)
                    : ConcreteDictionaryConverter(value_type, options, pool),
                      decoder_(value_type, options_) {}

            turbo::Result<std::shared_ptr<Array>> Convert(const BlockParser &parser,
                                                   int32_t col_index) override {
                // We use a fixed index width so that all column chunks get the same index type
                using BuilderType = Dictionary32Builder<T>;
                using value_type = typename ValueDecoderType::value_type;

                BuilderType builder(value_type_, pool_);
                TURBO_RETURN_NOT_OK(PresizeBuilder(parser, &builder));

                auto visit = [&](const uint8_t *data, uint32_t size, bool quoted) -> turbo::Status {
                    if (decoder_.is_null(data, size, quoted /* quoted */)) {
                        return builder.append_null();
                    }
                    if (TURBO_UNLIKELY(builder.dictionary_length() > max_cardinality_)) {
                        return turbo::out_of_range_error("Dictionary length exceeded max cardinality");
                    }
                    value_type value{};
                    TURBO_RETURN_NOT_OK(decoder_.Decode(data, size, quoted, &value));
                    return builder.append(value);
                };
                TURBO_RETURN_NOT_OK(parser.visit_column(col_index, visit));

                std::shared_ptr<Array> res;
                TURBO_RETURN_NOT_OK(builder.finish(&res));
                return res;
            }

            void SetMaxCardinality(int32_t max_length) override { max_cardinality_ = max_length; }

        protected:
            turbo::Status Initialize() override {
                util::InitializeUTF8();
                return decoder_.Initialize();
            }

            ValueDecoderType decoder_;
            int32_t max_cardinality_ = std::numeric_limits<int32_t>::max();
        };

//
// Concrete Converter factory for timestamps
//

        template<template<typename, typename> class ConverterType>
        std::shared_ptr<Converter> MakeTimestampConverter(const std::shared_ptr<DataType> &type,
                                                          const ConvertOptions &options,
                                                          MemoryPool *pool) {
            if (options.timestamp_parsers.size() == 0) {
                // Default to ISO-8601
                return std::make_shared<ConverterType<TimestampType, InlineISO8601ValueDecoder>>(
                        type, options, pool);
            } else if (options.timestamp_parsers.size() == 1) {
                // Single user-supplied converter
                return std::make_shared<
                        ConverterType<TimestampType, SingleParserTimestampValueDecoder>>(type, options,
                                                                                         pool);
            } else {
                // Multiple converters, must iterate for each value
                return std::make_shared<
                        ConverterType<TimestampType, MultipleParsersTimestampValueDecoder>>(type, options,
                                                                                            pool);
            }
        }

//
// Concrete Converter factory for reals
//

        template<typename ConverterType, template<typename...> class ConcreteConverterType,
                typename Type, typename DecoderType>
        std::shared_ptr<ConverterType> MakeRealConverter(const std::shared_ptr<DataType> &type,
                                                         const ConvertOptions &options,
                                                         MemoryPool *pool) {
            if (options.decimal_point == '.') {
                return std::make_shared<ConcreteConverterType<Type, DecoderType>>(type, options,
                                                                                  pool);
            }
            return std::make_shared<
                    ConcreteConverterType<Type, CustomDecimalPointValueDecoder<DecoderType>>>(
                    type, options, pool);
        }

    }  // namespace

/////////////////////////////////////////////////////////////////////////
// Base Converter class implementation

    Converter::Converter(const std::shared_ptr<DataType> &type, const ConvertOptions &options,
                         MemoryPool *pool)
            : options_(options), pool_(pool), type_(type) {}

    DictionaryConverter::DictionaryConverter(const std::shared_ptr<DataType> &value_type,
                                             const ConvertOptions &options, MemoryPool *pool)
            : Converter(dictionary(int32(), value_type), options, pool),
              value_type_(value_type) {}

    turbo::Result<std::shared_ptr<Converter>> Converter::create(const std::shared_ptr<DataType> &type,
                                                       const ConvertOptions &options,
                                                       MemoryPool *pool) {
        std::shared_ptr<Converter> ptr;

        switch (type->id()) {
#define CONVERTER_CASE(TYPE_ID, CONVERTER_TYPE)         \
  case TYPE_ID:                                         \
    ptr.reset(new CONVERTER_TYPE(type, options, pool)); \
    break;

#define NUMERIC_CONVERTER_CASE(TYPE_ID, TYPE_CLASS) \
  CONVERTER_CASE(TYPE_ID,                           \
                 (PrimitiveConverter<TYPE_CLASS, NumericValueDecoder<TYPE_CLASS>>))

#define REAL_CONVERTER_CASE(TYPE_ID, TYPE_CLASS, DECODER)                        \
  case TYPE_ID:                                                                  \
    ptr = MakeRealConverter<Converter, PrimitiveConverter, TYPE_CLASS, DECODER>( \
        type, options, pool);                                                    \
    break;

            CONVERTER_CASE(Type::NA, NullConverter)
            NUMERIC_CONVERTER_CASE(Type::INT8, Int8Type)
            NUMERIC_CONVERTER_CASE(Type::INT16, Int16Type)
            NUMERIC_CONVERTER_CASE(Type::INT32, Int32Type)
            NUMERIC_CONVERTER_CASE(Type::INT64, Int64Type)
            NUMERIC_CONVERTER_CASE(Type::UINT8, UInt8Type)
            NUMERIC_CONVERTER_CASE(Type::UINT16, UInt16Type)
            NUMERIC_CONVERTER_CASE(Type::UINT32, UInt32Type)
            NUMERIC_CONVERTER_CASE(Type::UINT64, UInt64Type)
            NUMERIC_CONVERTER_CASE(Type::FP32, Fp32Type)
            NUMERIC_CONVERTER_CASE(Type::FP64, Fp64Type)
            REAL_CONVERTER_CASE(Type::DECIMAL, Decimal128Type, DecimalValueDecoder)
            NUMERIC_CONVERTER_CASE(Type::DATE32, Date32Type)
            NUMERIC_CONVERTER_CASE(Type::DATE64, Date64Type)
            NUMERIC_CONVERTER_CASE(Type::TIME32, Time32Type)
            NUMERIC_CONVERTER_CASE(Type::TIME64, Time64Type)
            CONVERTER_CASE(Type::BOOL, (PrimitiveConverter<BooleanType, BooleanValueDecoder>))
            CONVERTER_CASE(Type::BINARY,
                           (PrimitiveConverter<BinaryType, BinaryValueDecoder<false>>))
            CONVERTER_CASE(Type::LARGE_BINARY,
                           (PrimitiveConverter<LargeBinaryType, BinaryValueDecoder<false>>))
            CONVERTER_CASE(Type::FIXED_SIZE_BINARY,
                           (PrimitiveConverter<FixedSizeBinaryType, FixedSizeBinaryValueDecoder>))

            case Type::TIMESTAMP:
                ptr = MakeTimestampConverter<PrimitiveConverter>(type, options, pool);
                break;

            case Type::STRING:
                if (options.check_utf8) {
                    ptr = std::make_shared<PrimitiveConverter<StringType, BinaryValueDecoder<true>>>(
                            type, options, pool);
                } else {
                    ptr = std::make_shared<PrimitiveConverter<StringType, BinaryValueDecoder<false>>>(
                            type, options, pool);
                }
                break;

            case Type::LARGE_STRING:
                if (options.check_utf8) {
                    ptr = std::make_shared<
                            PrimitiveConverter<LargeStringType, BinaryValueDecoder<true>>>(type, options,
                                                                                           pool);
                } else {
                    ptr = std::make_shared<
                            PrimitiveConverter<LargeStringType, BinaryValueDecoder<false>>>(type, options,
                                                                                            pool);
                }
                break;

            case Type::DICTIONARY: {
                const auto &dict_type = turbo::checked_cast<const DictionaryType &>(*type);
                if (dict_type.index_type()->id() != Type::INT32) {
                    return turbo::unimplemented_error(
                            "CSV conversion to dictionary only supported for int32 indices, "
                            "got ",
                            type->to_string());
                }
                return DictionaryConverter::create(dict_type.get_value_type(), options, pool);
            }

            default: {
                return turbo::unimplemented_error("CSV conversion to ", type->to_string(),
                                              " is not supported");
            }

#undef CONVERTER_CASE
#undef NUMERIC_CONVERTER_CASE
#undef REAL_CONVERTER_CASE
        }
        TURBO_RETURN_NOT_OK(ptr->Initialize());
        return ptr;
    }

    turbo::Result<std::shared_ptr<DictionaryConverter>> DictionaryConverter::create(
            const std::shared_ptr<DataType> &type, const ConvertOptions &options,
            MemoryPool *pool) {
        std::shared_ptr<DictionaryConverter> ptr;

        switch (type->id()) {
#define CONVERTER_CASE(TYPE_ID, TYPE, VALUE_DECODER_TYPE)                             \
  case TYPE_ID:                                                                       \
    ptr.reset(                                                                        \
        new TypedDictionaryConverter<TYPE, VALUE_DECODER_TYPE>(type, options, pool)); \
    break;

#define REAL_CONVERTER_CASE(TYPE_ID, TYPE_CLASS, DECODER)                              \
  case TYPE_ID:                                                                        \
    ptr = MakeRealConverter<DictionaryConverter, TypedDictionaryConverter, TYPE_CLASS, \
                            DECODER>(type, options, pool);                             \
    break;

            // XXX Are 32-bit types useful?
            CONVERTER_CASE(Type::INT32, Int32Type, NumericValueDecoder<Int32Type>)
            CONVERTER_CASE(Type::INT64, Int64Type, NumericValueDecoder<Int64Type>)
            CONVERTER_CASE(Type::UINT32, UInt32Type, NumericValueDecoder<UInt32Type>)
            CONVERTER_CASE(Type::UINT64, UInt64Type, NumericValueDecoder<UInt64Type>)
            CONVERTER_CASE(Type::FP32, Fp32Type, NumericValueDecoder<Fp32Type>)
            CONVERTER_CASE(Type::FP64, Fp64Type, NumericValueDecoder<Fp64Type>)
            REAL_CONVERTER_CASE(Type::DECIMAL, Decimal128Type, DecimalValueDecoder)
            CONVERTER_CASE(Type::FIXED_SIZE_BINARY, FixedSizeBinaryType,
                           FixedSizeBinaryValueDecoder)
            CONVERTER_CASE(Type::BINARY, BinaryType, BinaryValueDecoder<false>)
            CONVERTER_CASE(Type::LARGE_BINARY, LargeBinaryType, BinaryValueDecoder<false>)

            case Type::STRING:
                if (options.check_utf8) {
                    ptr = std::make_shared<
                            TypedDictionaryConverter<StringType, BinaryValueDecoder<true>>>(type, options,
                                                                                            pool);
                } else {
                    ptr = std::make_shared<
                            TypedDictionaryConverter<StringType, BinaryValueDecoder<false>>>(
                            type, options, pool);
                }
                break;

            case Type::LARGE_STRING:
                if (options.check_utf8) {
                    ptr = std::make_shared<
                            TypedDictionaryConverter<LargeStringType, BinaryValueDecoder<true>>>(
                            type, options, pool);
                } else {
                    ptr = std::make_shared<
                            TypedDictionaryConverter<LargeStringType, BinaryValueDecoder<false>>>(
                            type, options, pool);
                }
                break;

            default: {
                return turbo::unimplemented_error("CSV dictionary conversion to ", type->to_string(),
                                              " is not supported");
            }

#undef CONVERTER_CASE
#undef REAL_CONVERTER_CASE
        }
        TURBO_RETURN_NOT_OK(ptr->Initialize());
        return ptr;
    }

}  // namespace nebula::csv
